/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.nifi.processors.huawei.obs;

import com.obs.services.ObsClient;
import com.obs.services.model.GetObjectMetadataRequest;
import com.obs.services.model.GetObjectRequest;
import com.obs.services.model.ObjectMetadata;
import com.obs.services.model.ObsObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.nifi.annotation.behavior.InputRequirement;
import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
import org.apache.nifi.annotation.behavior.SupportsBatching;
import org.apache.nifi.annotation.behavior.WritesAttribute;
import org.apache.nifi.annotation.behavior.WritesAttributes;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.components.ConfigVerificationResult;
import org.apache.nifi.components.ConfigVerificationResult.Outcome;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.expression.ExpressionLanguageScope;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.flowfile.attributes.CoreAttributes;
import org.apache.nifi.logging.ComponentLog;
import org.apache.nifi.processor.DataUnit;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.processors.huawei.abstractprocessor.AbstractOBSProcessor;

import java.io.IOException;
import java.net.URLDecoder;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import static org.apache.nifi.processors.huawei.common.PropertyDescriptors.*;
import static org.apache.nifi.processors.huawei.obs.Constants.*;

@SupportsBatching
@InputRequirement(Requirement.INPUT_REQUIRED)
@Tags({"HuaweiCloud", "OBS",  "Get", "Fetch"})
@CapabilityDescription("Retrieves the contents of an OBS Object and writes it to the content of a FlowFile")
@WritesAttributes({
    @WritesAttribute(attribute = OBS_BUCKET, description = "The name of the OBS bucket"),
    @WritesAttribute(attribute = OBS_OPERATION, description = "The name of the operator"),
    @WritesAttribute(attribute = "path", description = "The path of the file"),
    @WritesAttribute(attribute = "absolute.path", description = "The path of the file"),
    @WritesAttribute(attribute = "filename", description = "The name of the file"),
    @WritesAttribute(attribute = "mime.type", description = "If OBS provides the content type/MIME type, this attribute will hold that file"),
    @WritesAttribute(attribute = OBS_E_TAG, description = "The ETag that can be used to see if the file has changed"),
    @WritesAttribute(attribute = OBS_EXCEPTION, description = "The class name of the exception thrown during processor execution"),
    @WritesAttribute(attribute = OBS_ADDITIONAL_DETAILS, description = "The OBS supplied detail from the failed operation"),
    @WritesAttribute(attribute = OBS_STATUS_CODE, description = "The HTTP error code (if available) from the failed operation"),
    @WritesAttribute(attribute = OBS_ERROR_CODE, description = "The OBS moniker of the failed operation"),
    @WritesAttribute(attribute = OBS_ERROR_Message, description = "The OBS exception message from the failed operation"),
    @WritesAttribute(attribute = OBS_EXPIRATION_TIME, description = "If the file has an expiration date, this attribute will be set, containing the milliseconds since epoch in UTC time"),
    @WritesAttribute(attribute = OBS_SSE_ALGORITHM, description = "The server side encryption algorithm of the object"),
    @WritesAttribute(attribute = OBS_VERSION, description = "The version of the OBS object"),
    @WritesAttribute(attribute = OBS_ENCRYPTION_STRATEGY, description = "The name of the encryption strategy that was used to store the OBS object (if it is encrypted)"),})
public class FetchOBSObject extends AbstractOBSProcessor {
    public static final PropertyDescriptor RANGE_START = new PropertyDescriptor.Builder()
            .name("range-start")
            .displayName("Range Start")
            .description("The byte position at which to start reading from the object. An empty value or a value of " +
                    "zero will start reading at the beginning of the object.")
            .addValidator(StandardValidators.DATA_SIZE_VALIDATOR)
            .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
            .required(false)
            .build();

    public static final PropertyDescriptor RANGE_LENGTH = new PropertyDescriptor.Builder()
            .name("range-length")
            .displayName("Range Length")
            .description("The number of bytes to download from the object, starting from the Range Start. An empty " +
                    "value or a value that extends beyond the end of the object will read to the end of the object.")
            .addValidator(StandardValidators.createDataSizeBoundsValidator(1, Long.MAX_VALUE))
            .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
            .required(false)
            .build();

    public static final List<PropertyDescriptor> properties = Collections.unmodifiableList(
            Arrays.asList(
                    OBS_REGION,
                    ENDPOINT_OVERRIDE_URL,
                    BUCKET,
                    KEY,
                    ACCESS_KEY,
                    SECRET_KEY,
                    HUAWEI_CREDENTIALS_PROVIDER_SERVICE,
                    ENCRYPTION_SERVICE,
                    TIMEOUT,
                    RANGE_START,
                    RANGE_LENGTH,
                    PROXY_CONFIGURATION_SERVICE));

    @Override
    public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
        return properties;
    }

    @Override
    public List<ConfigVerificationResult> verify(ProcessContext context, ComponentLog verificationLogger, Map<String, String> attributes) {
        final List<ConfigVerificationResult> results = new ArrayList<>(super.verify(context, verificationLogger, attributes));

        final String bucket = context.getProperty(BUCKET).evaluateAttributeExpressions(attributes).getValue();
        final String key = context.getProperty(KEY).evaluateAttributeExpressions(attributes).getValue();

        final ObsClient client =  OBSUtils.createClient(context, getLogger());
        final GetObjectMetadataRequest request = createGetObjectMetadataRequest(context, attributes);

        try {
            final ObjectMetadata objectMetadata = client.getObjectMetadata(request);
            final long byteCount = objectMetadata.getContentLength();
            results.add(new ConfigVerificationResult.Builder()
                    .verificationStepName("HEAD OBS Object")
                    .outcome(Outcome.SUCCESSFUL)
                    .explanation(String.format("Successfully performed HEAD on [%s] (%s bytes) from Bucket [%s]", key, byteCount, bucket))
                    .build());
        } catch (final Exception e) {
            getLogger().error(String.format("Failed to fetch [%s] from Bucket [%s]", key, bucket), e);
            results.add(new ConfigVerificationResult.Builder()
                    .verificationStepName("HEAD OBS Object")
                    .outcome(Outcome.FAILED)
                    .explanation(String.format("Failed to perform HEAD on [%s] from Bucket [%s]: %s", key, bucket, e.getMessage()))
                    .build());
        }

        return results;
    }

    @Override
    public void onTrigger(final ProcessContext context, final ProcessSession session) {
        getLogger().warn("FetchObs Trigger {}", System.currentTimeMillis());
        FlowFile flowFile = session.get();
        if (flowFile == null) {
            return;
        }

        final long startNanos = System.nanoTime();
        final Map<String, Object> attributes = new HashMap<>();

        final String bucket = context.getProperty(BUCKET).evaluateAttributeExpressions(flowFile).getValue();
        final String key = context.getProperty(KEY).evaluateAttributeExpressions(flowFile).getValue();
        final String region = context.getProperty(OBS_REGION).getValue();
        final String endpoint = context.getProperty(ENDPOINT_OVERRIDE_URL).getValue();

        final ObsClient client = getClient();
        final GetObjectRequest request = createGetObjectRequest(context, flowFile.getAttributes(), attributes);
        try {
            final ObsObject obsObject = client.getObject(request);
            if (obsObject == null) {
                throw new IOException("HuaweiCloud refused to execute this request.");
            }
            flowFile = session.importFrom(obsObject.getObjectContent(), flowFile);
            attributes.put(OBS_OBJECT, key);
            attributes.put(FILENAME, obsObject.getObjectKey());
            attributes.put(OBS_OPERATION, "FETCH");
            attributes.put(OBS_BUCKET, obsObject.getBucketName());
            final ObjectMetadata metadata = obsObject.getMetadata();
            if (StringUtils.isNotBlank(metadata.getContentDisposition())) {
                final String contentDisposition = URLDecoder.decode(metadata.getContentDisposition(), StandardCharsets.UTF_8.name());
                if (contentDisposition.equals(PutOBSObject.CONTENT_DISPOSITION_INLINE) || contentDisposition.startsWith(CONTENT_DISPOSITION_ATTACHMENT)) {
                    setFilePathAttributes(attributes, key);
                } else {
                    setFilePathAttributes(attributes, contentDisposition);
                }
            }
            if (metadata.getContentMd5() != null) {
                attributes.put(HASH_VALUE, metadata.getContentMd5());
                attributes.put(HASH_ALGORITHM, "MD5");
            }
            if (metadata.getContentType() != null) {
                attributes.put(CoreAttributes.MIME_TYPE.key(), metadata.getContentType());
            }
            if (metadata.getEtag() != null) {
                attributes.put(OBS_E_TAG, metadata.getEtag());
            }
            if (metadata.getExpires() != null) {
                attributes.put(OBS_EXPIRATION_TIME, String.valueOf(metadata.getExpires()));
            }
            if (metadata.getAllMetadata() != null) {
                attributes.putAll(metadata.getAllMetadata());
            }
        } catch (Exception ioe) {
            flowFile = extractExceptionDetails(ioe, session, flowFile);
            getLogger().error("Failed to retrieve OBS Object for {}; routing to failure", flowFile, ioe);
            flowFile = session.penalize(flowFile);
            session.transfer(flowFile, REL_FAILURE);
            return;
        }
        if (!attributes.isEmpty()) {
            flowFile = session.putAllAttributes(flowFile, attributes.entrySet().stream().
                    collect(Collectors.toMap(item -> String.valueOf(item.getKey()) ,
                            val ->  String.valueOf(val.getValue()))));
        }

        session.transfer(flowFile, REL_SUCCESS);
        final long transferMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
        getLogger().info("Successfully retrieved OBS Object for {} in {} millis; routing to success", new Object[]{flowFile, transferMillis});
        session.getProvenanceReporter().fetch(flowFile, OBSUtils.getUrl(region, endpoint, bucket, key), transferMillis);
    }

    private GetObjectMetadataRequest createGetObjectMetadataRequest(final ProcessContext context, final Map<String, String> attributes) {
        final String bucket = context.getProperty(BUCKET).evaluateAttributeExpressions(attributes).getValue();
        final String key = context.getProperty(KEY).evaluateAttributeExpressions(attributes).getValue();
        return new GetObjectMetadataRequest(bucket, key);
    }

    private GetObjectRequest createGetObjectRequest(final ProcessContext context, final Map<String, String> flowFileAttributes, Map<String, Object> attributes) {
        final String bucket = context.getProperty(BUCKET).evaluateAttributeExpressions(flowFileAttributes).getValue();
        final String key = context.getProperty(KEY).evaluateAttributeExpressions(flowFileAttributes).getValue();
        final long rangeStart = (context.getProperty(RANGE_START).isSet() ? context.getProperty(RANGE_START).evaluateAttributeExpressions(flowFileAttributes).asDataSize(DataUnit.B).longValue() : 0L);
        final Long rangeLength = (context.getProperty(RANGE_LENGTH).isSet() ? context.getProperty(RANGE_LENGTH).evaluateAttributeExpressions(flowFileAttributes).asDataSize(DataUnit.B).longValue() : null);
        final GetObjectRequest request = new GetObjectRequest(bucket, key);

        // Since the effect of the byte range 0- is equivalent to not sending a
        // byte range and works for both zero and non-zero length objects,
        // the single argument setRange() only needs to be called when the
        // first byte position is greater than zero.
        if (rangeLength != null) {
            request.setRangeStart(rangeStart);
            request.setRangeEnd(rangeStart + rangeLength - 1);
        } else if (rangeStart > 0) {
            request.setRangeStart(rangeStart);
        }

        final ObsServiceEncryptionService encryptionService = context.getProperty(ENCRYPTION_SERVICE).asControllerService(ObsServiceEncryptionService.class);
        if (encryptionService != null) {
            encryptionService.configureGetObjectRequest(request, new ObjectMetadata());
        }

        if (request.getSseCHeader() !=null && request.getSseCHeader().getSSEAlgorithm() != null) {
            String sseAlgorithmName = request.getSseCHeader().getSSEAlgorithm().name();
            attributes.put(OBS_SSE_ALGORITHM, sseAlgorithmName);
            attributes.put(OBS_ENCRYPTION_STRATEGY, ObsServiceEncryptionService.STRATEGY_NAME_SSE_C);
        }
        return request;
    }

    protected void setFilePathAttributes(Map<String, Object> attributes, String filePathName) {
        final int lastSlash = filePathName.lastIndexOf("/");
        if (lastSlash > -1 && lastSlash < filePathName.length() - 1) {
            attributes.put(CoreAttributes.PATH.key(), filePathName.substring(0, lastSlash));
            attributes.put(CoreAttributes.ABSOLUTE_PATH.key(), filePathName);
            attributes.put(CoreAttributes.FILENAME.key(), filePathName.substring(lastSlash + 1));
        } else {
            attributes.put(CoreAttributes.FILENAME.key(), filePathName);
        }
    }
}
